# -*- coding: utf-8 -*-
"""
Created on Wed May  8 15:34:20 2019

@author: nchji
"""
from copy import deepcopy
from jili.core.printlog import print
import threading
import pickle,zmq,time
import os
from concurrent.futures import ThreadPoolExecutor
from functools import partial
print("Version:20211103")
def Singleton(cls):
    _instance = {}
    def _singleton(*args, **kargs):
        if cls not in _instance:
            _instance[cls] = cls(*args, **kargs)
        return _instance[cls]

    return _singleton
class SingletonType(type):
    _instance_lock = threading.Lock()
    def __call__(cls):
        if not hasattr(cls, "_instance"):
            with SingletonType._instance_lock:
                if not hasattr(cls, "_instance"):
                    cls._instance = super(SingletonType,cls).__call__()
        return cls._instance
class Counter(metaclass=SingletonType):
    def __init__(self,init=0):
        self.number=init
    def getid(self):
        self.number=self.number+1
        return self.number

counter=Counter()
sourceid=os.getpid()
def tobytes(msg):
    if not isinstance(msg,bytes):
        return bytes(msg,encoding='utf8')
    else:
        return msg
def tostring(msg):
    if isinstance(msg,str):
        return msg
    else:
        return str(msg,encoding='utf8')
def getsource(isreply=False):
    t={}
    t["pid"]=sourceid
    t["seqid"]=counter.getid()
    t["isreply"]=isreply
    return t
def encodemsg(msgtype,subject="",msg={},isreply=False,s=""):
    if s=="":
        t=getsource(isreply)
        data=(t,msg)
        s0=tobytes(msgtype)+b"|"+tobytes(subject)+b"|"+pickle.dumps(data)
        return s0,t["seqid"]
    else:
        t=getsource()
        data = (t, msg)
        s0=tobytes(msgtype)+b"|"+tobytes(subject)+b"|"+pickle.dumps(data)
        return s0,t["seqid"]
        #return zlib.compress(s),t["seqid"]
    #self.__end_send.send(zlib.compress((s)
def decodemsg(s):
    #ms=zlib.decompress(s)
    # print(s)
    ms=s.split(b'|')
    m={}
    if len(ms)==3:
        m["0"]=tostring(ms[0])
        m["1"]=tostring(ms[1])
        data=pickle.loads(ms[2])
        m["s"]=data[0]
        m["2"]=data[1]
    elif len(ms)>3:
        m["0"]=tostring(ms[0])
        m["1"]=tostring(ms[1])
        ms0 = ms[2]
        for i in ms[3:]:
            ms0 = ms0+b"|"+i
        data = pickle.loads(ms0)
        m["s"] = data[0]
        m["2"] = data[1]
    else:
        print("error","decodemsg:len",len(ms),s)
    return m
# mohe_boxs={}
@Singleton
class Mohe_Recv:
    def __init__(self):
        self.__context_recv = zmq.Context()
        self.__end_recv = self.__context_recv.socket(zmq.SUB)
        self.__end_recv.setsockopt(zmq.TCP_KEEPALIVE, 1)
        self.__end_recv.setsockopt(zmq.TCP_KEEPALIVE_INTVL, 1)
        self.__end_recv.setsockopt(zmq.TCP_KEEPALIVE_IDLE, 60)
        self.__end_recv.setsockopt(zmq.TCP_KEEPALIVE_CNT, 5)
        self.__end_sends=[]
        self.recv_addrrs=[]
        self.__onfunclist = {}
        self.__acknotify = {}
        self.__acknotify_sycn = {}
        self.msgseqkey = {}
        self.max_workers=1
        self.executor = ThreadPoolExecutor(max_workers=1)
        self.subscribe(self.ontinitmsgseqkey, msgtype="initevent", subject="")
        self.subscribe(self.onacknotify, msgtype="acknotify", subject=str(sourceid))
        self.isnoblock = False
        self.isstoprecv = False
        self.isrunning = False
        self.t1 = None
        self.debug=False
    def add_addr(self,recv_addrr,send):
        if recv_addrr not in self.recv_addrrs:
            self.recv_addrrs.append(recv_addrr)
            self.__end_recv.connect(recv_addrr)
            self.__end_sends.append(send)
    def setmax_workers(self,maxwokers):
        self.maxwokers = maxwokers
    def close(self):
        self.__end_recv.close()
        self.__context_recv.destroy()

    def stop(self):
        self.isrunning = False
        self.isstoprecv = True

    def run_recv(self,isnoblock=False,maxwokers=None):
        if not self.isrunning:
            if maxwokers:
                self.maxwokers=maxwokers
            # self.debug=log.level
            self.executor = ThreadPoolExecutor(max_workers=self.maxwokers)
            self.isnoblock =isnoblock
            #print("runrecv", "isnoblock", self.isnoblock, "")
            self.t1 = threading.Thread(target=self.__deal, args=())
            self.t1.start()
            print("runrecv", self.t1.getName(), "isnoblock:", self.isnoblock)
            self.isstoprecv = False
            self.isrunning = True

    def isActive(self):
        return self.__end_recv.isalive()
    def subscribe(self,onfunc,msgtype="",subject=""):
        rst=()
        if subject=="":
            s = tobytes(msgtype) + b"|"
        else:
            s=tobytes(msgtype)+b"|"+tobytes(subject)+b"|"
        if msgtype not in self.__onfunclist.keys():
            self.__onfunclist[msgtype]={}
        if subject not in self.__onfunclist[msgtype].keys():
            self.__onfunclist[msgtype][subject]=[]
        if onfunc not in self.__onfunclist[msgtype][subject]:  # 去重多次注册函数，否则会多次
            self.__onfunclist[msgtype][subject].append(onfunc)
            self.__end_recv.subscribe(s)
        rst = (s, self.__onfunclist[msgtype][subject])
        return rst
    def unsubscribe(self,msgtype="",subject="",onfunc=""):
        if subject!="" :
            s=tobytes(msgtype)+b"|"+tobytes(subject)+b"|"
            if msgtype in self.__onfunclist.keys():
                if subject in self.__onfunclist[msgtype].keys():
                    if onfunc=="":
                        del self.__onfunclist[msgtype][subject]
                        self.__end_recv.unsubscribe(s)
                    else:
                        self.__onfunclist[msgtype][subject].remove(onfunc)
                        if len(self.__onfunclist[msgtype][subject])==0:
                            del self.__onfunclist[msgtype][subject]
                            self.__end_recv.unsubscribe(s)
        else:
            s=tobytes(msgtype)+b"|"
            if msgtype in self.__onfunclist.keys():
                if onfunc=="":
                    del self.__onfunclist[msgtype]
                    self.__end_recv.unsubscribe(s)
                else:
                    if subject in self.__onfunclist[msgtype].keys():
                        self.__onfunclist[msgtype][subject].remove(onfunc)
                        if len(self.__onfunclist[msgtype][subject])==0:
                            del self.__onfunclist[msgtype][subject]
                            self.__end_recv.unsubscribe(s)
        return (s,self.__onfunclist)
    def isunduplicate(self, s):
        pid = s["pid"]
        seqid = s["seqid"]
        if pid in self.msgseqkey.keys():
            if seqid > self.msgseqkey[pid]:
                return True
            else:
                return False
        else:
            return True

    def ontinitmsgseqkey(self, msg):
        pid = msg["s"]["pid"]
        if pid in self.msgseqkey.keys():
            del self.msgseqkey[pid]
    def set_acknotify_sync(self,seqid, t):
        self.__acknotify[seqid]=t
    def set_acknotify(self,seqid, acknotify):
        self.__acknotify[seqid]=acknotify
    def onacknotify(self, msg):
        seqid = msg["s"]["seqid"]
        msg["e"] = msg["2"]
        if seqid in self.__acknotify.keys():
            call=self.__acknotify.pop(seqid)
            call(msg)
        elif seqid in self.__acknotify_sycn.keys():
            t=self.__acknotify_sycn.pop(seqid)
            t.update(msg)
            t.set()
    def __impl_send(self,msg):
        for i in self.__end_sends:
            i.send(msg)
    def acksendreply(self,msg):
        t=msg["s"]
        t["isreply"]=False
        mstr,_=encodemsg("acknotify",str(msg["s"]["pid"]),msg={},isreply=False,s=t)
        self.__impl_send(mstr)
    def getsubcall(self):
        return self.__onfunclist

    def __deal(self):
        if self.isnoblock:
            self.__deal_noblock()
        else:
            self.__deal_block()

    def trycall(self, call, args):
        try:
            call(args)
        except Exception as e:
            print("error", "trycall:", call, e)
            print("error", "data", args)

    def __deal_block(self):
        while self.isrunning:
            try:
                m = self.__end_recv.recv()
                print("debug", "block.m:", m)
                msg = decodemsg(m)
                print("debug", "decodemsg", msg)
                if msg:
                    msgtype = msg["0"]
                    subject = msg["1"]
                    if msgtype in self.__onfunclist.keys():
                        if "" in self.__onfunclist[msgtype].keys():
                            for call in self.__onfunclist[msgtype][""]:
                                call0 = partial(self.trycall, call)
                                if "s" in msg.keys():
                                    s = msg["s"]
                                    if s["isreply"]:
                                        self.acksendreply(msg)
                                        if self.isunduplicate(s):
                                            self.executor.submit(call0, (msg))
                                    else:
                                        self.executor.submit(call0, (msg))
                                else:
                                    self.executor.submit(call0, (msg))
                        if subject in self.__onfunclist[msgtype].keys():
                            for call in self.__onfunclist[msgtype][subject]:
                                call0 = partial(self.trycall, call)
                                if "s" in msg.keys():
                                    s = msg["s"]
                                    if s["isreply"]:
                                        self.acksendreply(msg)
                                        if self.isunduplicate(s):
                                            self.executor.submit(call0, (msg))
                                    else:
                                        self.executor.submit(call0, (msg))
                                else:
                                    self.executor.submit(call0, (msg))
                if self.isstoprecv:
                    break
            except Exception as e:
                print("error", "eventlib __deal:", e)
                if self.debug:
                    raise e
                if self.isstoprecv:
                    break

    def __deal_noblock(self):
        while self.isrunning:
            try:
                m = self.__end_recv.recv(flags=zmq.NOBLOCK)
                print("debug", "noblock.m:", m)
                if m:
                    msg = decodemsg(m)
                    # print("decodemsg",msg)
                    if msg:
                        msgtype = msg["0"]
                        subject = msg["1"]
                        if msgtype in self.__onfunclist.keys():
                            if "" in self.__onfunclist[msgtype].keys():
                                for call in self.__onfunclist[msgtype][""]:
                                    call0 = partial(self.trycall, call)
                                    if "s" in msg.keys():
                                        s = msg["s"]
                                        if s["isreply"]:
                                            self.acksendreply(msg)
                                            if self.isunduplicate(s):
                                                self.executor.submit(call0, (msg))
                                        else:
                                            self.executor.submit(call0, (msg))
                                    else:
                                        self.executor.submit(call0, (msg))
                            if subject in self.__onfunclist[msgtype].keys():
                                for call in self.__onfunclist[msgtype][subject]:
                                    call0 = partial(self.trycall, call)
                                    if "s" in msg.keys():
                                        s = msg["s"]
                                        if s["isreply"]:
                                            self.acksendreply(msg)
                                            if self.isunduplicate(s):
                                                self.executor.submit(call0, (msg))
                                        else:
                                            self.executor.submit(call0, (msg))
                                            # print("submit")
                                            # call(msg)
                                    else:
                                        self.executor.submit(call0, (msg))
                if self.isstoprecv:
                    break
            except zmq.ZMQError as e:
                if self.isstoprecv:
                    break
                time.sleep(0.01)
            except Exception as e:
                print("error", "eventlib __deal:", e)
                print("error", "msg", m)
                if self.debug:
                    raise e
                if self.isstoprecv:
                    break

mohe_recv=Mohe_Recv()

class mohe():
    def __init__(self,recv_addrr='tcp://127.0.0.1:9114',send_addrr='tcp://127.0.0.1:9115',maxwokers=2,isautorunrecv=True,isnoblock=False):
        global mohe_recv
        self.__end_recv = mohe_recv
        self.__end_recv.setmax_workers(maxwokers)
        self.__context_send = zmq.Context()
        self.__end_send = self.__context_send.socket(zmq.PUB)
        self.send_addrr=send_addrr
        if send_addrr:
            self.__end_send.connect(send_addrr)
        self.initevent()
        self.isnoblock=isnoblock
        self.isstoprecv=False
        self.isrunning = True
        self.t1=None
        if recv_addrr:
            self.__end_recv.add_addr(recv_addrr,self.__end_send)
            if isautorunrecv:
                print("auto run_recv")
                self.__end_recv.run_recv(isnoblock=self.isnoblock)
        time.sleep(0.1)
    def close(self):
        self.__end_recv.close()
        self.__end_send.close()
        self.__context_send.destroy()
    def stop(self):
        self.__end_recv.stop()
    def start(self):
        if self.t1:
            self.t1.start()
    def join(self):
        if self.t1:
            self.t1.join()
    def run(self):
        print("runrecv isnoblock_main", "isnoblock", self.isnoblock, "")
        self.__deal()
    def isActive(self):
        return (self.__end_send.TCP_KEEPALIVE,self.__end_recv.isActive())
    def setmax_workers(self,maxwokers):
        self.__end_recv.setmax_workers(maxwokers)
    def subscribe(self,onfunc,msgtype="",subject=""):
        rst = self.__end_recv.subscribe(onfunc,msgtype,subject)
        return rst
    def unsubscribe(self,msgtype="",subject="",onfunc=""):
        rst=self.__end_recv.unsubscribe(onfunc,msgtype,subject)
        return rst
    def __impl_send(self,mstr):
        self.__end_send.send(mstr)
    def initevent(self):
        mstr,_=encodemsg("initevent","")
        self.__impl_send(mstr)
    def safesend(self,msgtype,subject,msg,acknotify):
        mstr,seqid=encodemsg(msgtype,subject,msg,isreply=True)
        self.__impl_send(mstr)
        self.__end_recv.set_acknotify(seqid,acknotify)
        return seqid
    def safesend_sync(self,msgtype,subject,msg,timeout=45):
        mstr,seqid=encodemsg(msgtype,subject,msg,isreply=True)
        t=dict(event = threading.Event())
        self.__end_recv.set_acknotify_sync(seqid,t)
        t["event"].clear()
        self.__impl_send(mstr)
        timeout0=timeout/3
        rst={}
        rst["errcode"]=0
        if t["event"].wait(timeout0):
            if "e" in t.keys():
                rst["info"]=t["e"]
            else:
                t = dict(event=threading.Event())
                self.__end_recv.set_acknotify_sync(seqid, t)
                t["event"].clear()
                self.__impl_send(mstr)
                if t["event"].wait(timeout0):
                    if "e" in t.keys():
                        rst["info"]=t["e"]
                    else:
                        t = dict(event=threading.Event())
                        self.__end_recv.set_acknotify_sync(seqid, t)
                        t["event"].clear()
                        self.__impl_send(mstr)
                        if t["event"].wait(timeout0):
                            if "e" in t.keys():
                                rst["info"]=t["e"]
                            else:
                                rst["errcode"]=1
                                rst["info"] = "超时"
                        else:
                            rst["errcode"] = 1
                            rst["info"] = "超时"
                else:
                    rst["errcode"] = 1
                    rst["info"] = "超时"
        else:
            rst["errcode"] = 1
            rst["info"]="超时"
        if rst["errcode"]==1:
            print("warning","消息超时：",msgtype,subject,msg,timeout)
        return rst

    def send(self,msgtype,subject,msg):
        mstr,seqid=encodemsg(msgtype,subject,msg,isreply=False)
        self.__impl_send(mstr)
        return seqid
    def getsubcall(self):
        return self.__end_recv.getsubcall()